package blue.datacenter;

import java.io.IOException;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.util.Date;
import java.util.HashMap;
import java.util.Iterator;
import java.util.Set;
import blue.logger.Publisher;
import blue.logger.JsonTransform;

public class ConnManager {
	// TODO Fix Me, Redundancy Code.
	boolean has_send_privilege_msg_ = false;
	Date slave_last_time_;
	private Selector selector_;
	private int id_ = 100000;
	private HashMap<Integer,ServerInterface> server_id_map_;
	//private HashMap<Integer,ConnectionInterface> slave_conn_id_map_;
	//private HashMap<String,ConnectionInterface> slave_conn_sn_map_;
	//private HashMap<Integer,ConnectionInterface> machine_conn_id_map_;
	private HashMap<String,ConnectionInterface> machine_conn_sn_map_;
	private HashMap<Integer,ConnectionInterface> conn_id_map_;
	ConnectionInterface active_slave_conn_;
	ConnectionInterface privilege_slave_conn_;
	Publisher pub_;
	
	
	public ConnManager(Publisher pub) throws IOException {
        // 鍒涘缓閫夋嫨鍣�
		pub_ = pub;
        selector_ = Selector.open();
        conn_id_map_ = new HashMap<Integer, ConnectionInterface>();
        server_id_map_ = new HashMap<Integer, ServerInterface>();
        machine_conn_sn_map_ = new HashMap<String,ConnectionInterface>();
	}

	public void addServerConn(ServerInterface conn) throws IOException {
        // Set channel non-blocking & register
		ServerSocketChannel ssock_channel = conn.getServerSockChannel();
		ssock_channel.configureBlocking(false);
		server_id_map_.put(id_, conn);
		ssock_channel.register(selector_, SelectionKey.OP_ACCEPT).attach(id_++);
	}

	public void addMachineConn(ConnectionInterface conn) throws IOException {
		SocketChannel sock_channel = conn.getSockChannel();
		conn.setId(id_);
		conn_id_map_.put(conn.getId(), conn);
		//machine_conn_id_map_.put(id_, conn);
		machine_conn_sn_map_.put(conn.getSN(), conn);
		//addConn_(sock_channel, conn.getId());
		addConn_(conn, conn.getId());
		id_++;
	}

	public void addSlaveConn(ConnectionInterface conn) throws IOException {
		// TODO fix me, it active the last slave.
		SocketChannel sock_channel = conn.getSockChannel();
		conn.setId(id_);
		conn_id_map_.put(conn.getId(), conn);
		//addConn_(sock_channel, conn.getId());
		addConn_(conn, conn.getId());
		id_++;
		
		if (privilege_slave_conn_==null) {
			// The first slave connection must be from web_server, so it has privilege.
			privilege_slave_conn_ = conn;
			conn.setTypePrivilege();
		}
	}
	
	public void run() {
        try {
            while (true) {
                selector_.select(); // 闃诲 鐩村埌鏈夊氨缁簨浠朵负姝�
                Set<SelectionKey> readySelectionKey = selector_.selectedKeys();
                Iterator<SelectionKey> it = readySelectionKey.iterator();
                while (it.hasNext()) {
                    SelectionKey key = it.next();
                    if (key.isAcceptable()) {
                    	// Accept slave connection & Add it to selector.
                        System.out.println("Server accept a connection....");
                        eventAccept(key);
                        // 涓嬮潰杩欑鍐欐硶鏄湁闂鐨� 涓嶅簲璇ュ湪serverSocketChannel涓婇潰娉ㄥ唽
                        /*
                         * serverSocketChannel.configureBlocking(false);
                         * serverSocketChannel.register(in_selector,
                         * SelectionKey.OP_READ);
                         * serverSocketChannel.register(in_selector,
                         * SelectionKey.OP_WRITE);
                         */
                    }
                    else if (key.isReadable()) {
                    	// Read data & parse it & send it to corresponding socket.
                        eventRead(key);
                    }else if (key.isWritable()) {

                    }else {
                    	throw new DCFatalException("This should never be arrived.");
                    }
                    it.remove(); 
                }
            }
        } catch (Exception e) {
            // TODO: handle exception
            System.out.println("Error - " + e.getMessage());
            e.printStackTrace();
            System.exit(-1);
        }
	}
	/*Event Function*/
	private void eventAccept(SelectionKey key) throws IOException {
		Integer id = (Integer) key.attachment();
		ServerInterface server_conn = server_id_map_.get(id);
		SlaveConn slave_conn = server_conn.accept();
		addSlaveConn(slave_conn);
		System.out.println("Add a slave connection<"+slave_conn.getIP()+":"+slave_conn.getPort()+"> to list.");
	}
	
	private void eventRead(SelectionKey key) throws Exception {

		Integer id = (Integer)key.attachment();
		ConnectionInterface conn = conn_id_map_.get(id);
		// Read data.
		String raw_msg = conn.read();
		if (raw_msg==null) {eventException(key);return;}
		slaveConnectionUpdate(conn);
		if (! isSlaveAvailable(conn)) {return;}

	    // Send to corresponding connection.
		if (conn.getType()==ConnectionType.MachineConn) {
			// Send msg to slave
			String ip = conn.getIP();
			String port = new String()+conn.getPort();
			Message msg = new Message(ip,port,raw_msg);
			String send_str = msg.toJSON();
			// it come from privilege command.
			if (has_send_privilege_msg_){
				// The last command is a privilege message, so this response to privilege connection.
				privilege_slave_conn_.write(send_str);
				SelectionKey send_key = (SelectionKey)privilege_slave_conn_.getAttach();
				send_key.interestOps(SelectionKey.OP_READ); //??????
				has_send_privilege_msg_ = false;
			}else if (active_slave_conn_!=null) {
				// it is just a normal command
				active_slave_conn_.write(send_str);
				SelectionKey send_key = (SelectionKey)active_slave_conn_.getAttach();
				send_key.interestOps(SelectionKey.OP_READ); //??????
				// need OP_READ
				System.out.println("Send to Slave > "+send_str);
			}
			else{
				System.out.println("Require a active slave_conn. Abandon message :"+raw_msg);
				return;
			}
			
			pubMsglog("1",false,msg);

		}else if(conn.getType()==ConnectionType.SlaveConn || conn.getType()==ConnectionType.PrivilegeSlaveConn) {
			// Parse&Send msg to machine
			String json_str = raw_msg;
			Message msg = new Message(json_str);
			String ip = msg.getSip();
			String port = msg.getPort();
			String machine_msg = msg.getData();
			String sn = ip+":"+port;
	
			ConnectionInterface send_conn = machine_conn_sn_map_.get(sn);
			if (send_conn!=null) {
				int write_cnt = send_conn.write(machine_msg);
				SelectionKey send_key = (SelectionKey)send_conn.getAttach();
				send_key.interestOps(SelectionKey.OP_READ); //??????
				System.out.println("Write count: "+write_cnt);
				System.out.println("Send to Machine ("+sn+") > "+machine_msg);
				slave_last_time_ = new Date();
				if (conn.getType()==ConnectionType.PrivilegeSlaveConn) has_send_privilege_msg_ = true;
			}else {
				System.out.println("Invalided target Machine "+sn);
				conn.write("Invalid IP&Port "+sn);
			}
			// Publish Message
			pubMsglog("1",true,msg);
			
		}else {
			System.out.println("This should never be arrived.");
			throw new Exception("This should never be arrived.");
		}
	}
	
	private void slaveConnectionUpdate(ConnectionInterface conn) {
		/**
		 * Whether this new connection can possess master.
		 */
		int clear_time = 20;
		if(conn.getType()==ConnectionType.SlaveConn) {
			Date curr_time = new Date();
			long diff_time = slave_last_time_!=null?curr_time.getTime()-slave_last_time_.getTime():0;
			if (active_slave_conn_==null || diff_time > clear_time) {
				active_slave_conn_ = conn;
			}
		}
	}
	
	private boolean isSlaveAvailable(ConnectionInterface conn) throws IOException {
		/**
		 *  Whether current slave have authority to pass.
		 */
		if(conn.getType()==ConnectionType.SlaveConn) {
			if (active_slave_conn_==conn) {
				return true;
			}else {
				conn.write("Master is busy");
				return false;
			}
		}
		return true;
	}
	
	private void eventException(SelectionKey key) throws DCFatalException, IOException {
		Integer id = (Integer)key.attachment();
		ConnectionInterface conn = conn_id_map_.get(id);	
	
		if (conn.getType()==ConnectionType.MachineConn) {
			// Broken Connection with machine, Reconnect it, re-register it.
			key.cancel();
			conn.reconnection();
			addConn_(conn, conn.getId());
			//throw new DCFatalException("Broken Connection with machine, Fatal Error.");
		}else if(conn.getType()==ConnectionType.SlaveConn) {
			// Remove connection.
			// TODO : ? need remove it from poller
			System.out.println("Clear a slave connection at "+conn.getIP()+":"+conn.getPort());
			conn_id_map_.remove(id);
			if (conn==active_slave_conn_) active_slave_conn_=null;
			key.cancel();  // Cancel listen for this key
		}else {
			throw new DCFatalException("This should never be arrived.");
		}
	}
	
	/*Other Function.*/
	private void pubMsglog(String id, boolean to_machine,Message msg) {
		// Publish Message
		String msg_id = id;
		String msg_send = null;
		String msg_recv = null;
		if (to_machine) {
			msg_send = "TM";
			msg_recv = msg.getSip()+":"+msg.getPort();;
		}else {
			msg_send = msg.getSip()+":"+msg.getPort();
			msg_recv = "TM";
		}
		String msg_data = msg.getData();
		MessageLog log = new MessageLog(msg_id,msg_send,msg_recv,msg_data);
		pub_.publish(JsonTransform.toJson(log));
	}
	
	private void addConn_(ConnectionInterface conn, int id) throws IOException {
		SocketChannel sock_channel = conn.getSockChannel();
		sock_channel.configureBlocking(false);
		SelectionKey key = sock_channel.register(selector_, SelectionKey.OP_READ);
		key.attach(id);
		conn.setAttach(key);
	}
}
